package com.wsm.executor.ExecutorPlatform;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @name: ExecutorPool
 * @Author: wangshimin
 * @Date: 2020/7/31 0031  上午 11:44
 * @Description:
 */
@Service
@ConditionalOnProperty(prefix = "executor", value = {"open"}, havingValue = "true")
public class Executor {

    @Autowired
    @Qualifier(value = "commonPool")
    private ThreadPoolExecutor COMMON_POOL;


    public <T> List<T> beginWork(List<IWrapper> wrappers) {
        return beginWork(0, COMMON_POOL, wrappers);
    }

    public <T> List<T> beginWork(long timeout, List<IWrapper> wrappers) {
        return beginWork(timeout, COMMON_POOL, wrappers);
    }

    public <T> List<T> beginWork(long timeout, IWrapper... wrappers) {
        return beginWork(timeout, COMMON_POOL, Arrays.stream(wrappers).collect(Collectors.toList()));
    }

    /**
     * 返回list集合
     *
     * @param timeout
     * @param pool
     * @param wrappers
     * @param <T>
     * @return
     */
    public <T> List<T> beginWork(long timeout, ThreadPoolExecutor pool, List<IWrapper> wrappers) {
        List<T> result = new CopyOnWriteArrayList<>();
        if (wrappers != null && wrappers.size() > 0) {
            CompletableFuture[] futureGos = new CompletableFuture[wrappers.size()];
            CompletableFuture[] futuresBacks = new CompletableFuture[wrappers.size()];
            for (int i = 0; i < wrappers.size(); i++) {
                IWrapper iWrapper = wrappers.get(i);
                CompletableFuture<T> futureGo = CompletableFuture.supplyAsync(() -> (T) iWrapper.getiWorker().action(iWrapper.getKey()), pool);
                futureGos[i] = futureGo;
                //接收线程执行完的返回值
                CompletableFuture<T> futureBack = futureGo.whenCompleteAsync((t, throwable) -> {
                    // 获取返回值
                    if (t != null) {
                        result.add(t);
                    }
                });
                futuresBacks[i] = futureBack;
            }
            CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(futuresBacks);
            if (!voidCompletableFuture.isDone()) {
                if (timeout == 0) {
                    try {
                        voidCompletableFuture.get();
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                        for (CompletableFuture future : futureGos) {
                            future.cancel(true);
                        }
                    }
                } else {
                    try {
                        voidCompletableFuture.get(timeout, TimeUnit.MILLISECONDS);
                    } catch (TimeoutException | InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                        for (CompletableFuture future : futureGos) {
                            future.cancel(true);
                        }
                    }
                }
            }
        }
        return result;
    }

    public <T, V> Map<T, V> beginWorkMap(List<IWrapper> wrappers) {
        return beginWorkMap(0, COMMON_POOL, wrappers);
    }

    public <T, V> Map<T, V> beginWorkMap(long timeout, List<IWrapper> wrappers) {
        return beginWorkMap(timeout, COMMON_POOL, wrappers);
    }

    public <T, V> Map<T, V> beginWorkMap(long timeout, IWrapper... wrappers) {
        return beginWorkMap(timeout, COMMON_POOL, Arrays.stream(wrappers).collect(Collectors.toList()));
    }

    /**
     * 返回map集合
     *
     * @param timeout
     * @param pool
     * @param wrappers
     * @param <T>
     * @param <V>
     * @return
     */
    public <T, V> Map<T, V> beginWorkMap(long timeout, ThreadPoolExecutor pool, List<IWrapper> wrappers) {
        Map<T, V> result = new ConcurrentHashMap<>();
        if (wrappers != null && wrappers.size() > 0) {
            CompletableFuture[] futureGos = new CompletableFuture[wrappers.size()];
            CompletableFuture[] futuresBacks = new CompletableFuture[wrappers.size()];
            for (int i = 0; i < wrappers.size(); i++) {
                IWrapper iWrapper = wrappers.get(i);
                CompletableFuture<T> futureGo = CompletableFuture.supplyAsync(() -> (T) iWrapper.getiWorker().action(iWrapper.getKey()), pool);
                futureGos[i] = futureGo;
                //接收线程执行完的返回值
                CompletableFuture<T> futureBack = futureGo.whenComplete((t, throwable) -> {
                    // 获取返回值
                    if (t != null) {
                        result.put((T) iWrapper.getKey(), (V) t);
                    }
                });
                futuresBacks[i] = futureBack;
            }
            CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(futuresBacks);
            if (timeout == 0) {
                voidCompletableFuture.join();
            } else {
                try {
                    voidCompletableFuture.get(timeout, TimeUnit.MILLISECONDS);
                } catch (TimeoutException | InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                    for (CompletableFuture future : futureGos) {
                        future.cancel(true);
                    }
                }
            }
        }
        return result;
    }

    public void shutDown() {
        COMMON_POOL.shutdown();
    }
}
